import java.io.*;
import java.net.*;
import java.sql.Timestamp;
import java.util.*;
import common.*;

public class MessagePasser extends Thread
{
	public static ArrayList<String> partnerIPs;
	public static ArrayList<Integer> partnerPorts;
	public static boolean [] alive;
	//The default port for incoming connections to this MessagePasser
	public final int PORT;
	//The handler for the process that receives an incoming message
	private Process parent;
	//The socket that will listen for incoming connections
	private ServerSocket incoming;
	//A List of all nodes currently connected to this node
	public SocketThread [] connections;
	private boolean running;
	
	/**
	* Constructor that builds a MessagePasser to listen for incoming connections and
	* pass messages on outgoing connections.
	* @param processMH A MessageHandler for the process using this MessagePasser, basicaly a callback function.
	* @param pID The Process ID of the process for this MessagePasser
	*/
	public MessagePasser(int _pID,Process _parent) throws IOException ////MessageHandler processMH, int pID) throws IOException
	{
		PORT = partnerPorts.get(_pID);
		parent = _parent;
		incoming = new ServerSocket(this.PORT);
		connections = new SocketThread [partnerIPs.size()];
		running = true;
		this.start();
	}
	/**
	* Listens for incoming socket connections to this node, and build a SocketThread when
	* one is received.
	*/
	public void run()
	{
		while(running)
		{
			//Try to accept a new connection
			try
			{
				System.out.println("Waiting for new connection on port " + incoming.getLocalPort());
				Socket newConnection = incoming.accept();
				System.out.println("Accepted new connection");
				//Dispatch a Thread for this connection and add it to our list
				new SocketThread(newConnection);
			}
			catch(SocketException e)
			{
				//This is normal behavior when a connection is closed.
				System.out.println("ServerSocket closed");
			}
			//If we cannot accept, then punt
			catch(IOException e)
			{
				System.out.println("Error accepting incoming connection: " + e);
			}
		}
	}
	/**
	* Receive an incoming message, checking for internal status messages.
	* @param msg The incoming message from the other node
	*/
	public void receive( Message msg )
	{
		// notify received
		Utils.print_log( "-postman: a msg[t=" + msg.get_type() + "] is received from proc(" + msg.get_sender_id() + ")" );		
		
		switch(msg.get_type())
		{	// network level msg handler
			case Def.PING:
				// ping from the router (or possibly the middleware)
				// need to respond that I'm alive
				
				Utils.print_log( "-postman(" + parent.myID + "): PING from pid=" + msg.get_sender_id() );
				break;
				
			case Def.DROP:
				// (id) disappears, remove the node (id) from the routing table  
				break;
				
			case Def.KILL:
				// preemptive KILL message from an authorized device. I'm supposed to die
				// do some graceful action, then suicide
				parent.shutdown();
				break;
				
			case Def.NOT_ASSIGNED:
				// dummy packet
				// shouldn't occur - I put this code for debugging purpose
				Utils.print_log( "ERROR: critical error (check your code @18rER#) -- MessagePasser.java" );
				break;
				
			default:
				parent.receive(msg);	// push the task should be handle at SW level
				break;
		}
		
	}
	/**
	* Send an outgoing message.
	* @param msg The outgoing message to Destination
	*/
	public boolean send(Message msg)
	{
		// either broadcasting or 1-to-1 comm.
		if( msg.get_receiver_id() == 0 ){	// broadcasting
			for( int receiver = 0; receiver < partnerIPs.size(); receiver++ ){

				// Check to see that we have a connection to that node
				if(addConnection(receiver))
				{
					Utils.print_log( "-postman(" + parent.myID + "): a msg[t=" + msg.get_type() + "] is sent to proc(" + msg.get_receiver_id() + ")" );
					connections[receiver].send(msg);
				}	
				return true;
			}
		}else{	// 1-to-1 comm
			int receiver = msg.get_receiver_id();
			
			//Check to see that we have a connection to that node
			if(addConnection(receiver))
			{
				Utils.print_log( "-postman(" + parent.myID + "): a msg[t=" + msg.get_type() + "] is sent to proc(" + msg.get_receiver_id() + ")" );
				return connections[receiver].send(msg);
			}
		}
		return false;
	}//end-of-send(msg)

	/**
	 * In the ME_TOKENLESS scheme, the processes will call this method to see if one got granted from all participants. 
	 * return true the parameter id_list has every ID in the network, otherwise false.
	 */
	 //TODO Complete this !!!!
	public boolean is_this_everyone( ArrayList<Integer> id_list ){
		Utils.print_log( "size(partner_list_mp) = " + partnerIPs.size() );
		Utils.print_log( "size(got_GRANTd_from) = " + id_list.size() );
		
		if( id_list.size() == partnerIPs.size() - 1 )
			return true;
		
		return false;
	}
		
	/**
	* Tries to add a connection to this MessagePasser
	* @param ip The destination ip address of this connection
	* @param port The port to connect with
	*/
	public boolean addConnection(int partnerID)
	{
		//If we are already connected to partner, just let the machine know
		if(connections[partnerID] != null)
		{
			return (alive[partnerID] = true);
		}
		//otherwaise try to establish the new connection
		else
		{
			try
			{
				Socket s = new Socket(partnerIPs.get(partnerID), partnerPorts.get(partnerID));
				new SocketThread(s, partnerID);
				return (alive[partnerID] = true);
			}
			catch(IOException e)
			{
				return (alive[partnerID] = false);
			}
		}
	}
	/**
	* Close down this message passer
	*/
	public void close() throws IOException
	{
		//Close all open connections
		for(SocketThread connection : connections)
		{
			connection.close();
		}
		incoming.close();
		//Stop accepting new connections
		running = false;
	}
	public static int getNetworkSize()
	{
		return partnerPorts.size();
	}
	public boolean isAlive(int partnerID)
	{
		return alive[partnerID];
	}	
	public String toString()
	{
		return Arrays.toString(connections);
	}
	
	/**
	* Inner class to handle all socket communication. Since multiple connections will
	* exist for each node, they each have to have multiple threads listening for incoming messages.
	**/
	private class SocketThread extends Thread
	{
		private Socket s;
		private boolean running;
		private ObjectInputStream sin;
		private ObjectOutputStream sout;
		/**
		* Constructor that builds a thread to listen for incoming messages on this socket
		* @param s The new socket to run on
		*/
		public SocketThread(Socket s) throws IOException
		{
			initialize(s);
		}
		/**
		* Constructor that builds a thread to listen for incoming messages on this socket
		* @param s The new socket to run on
		* @param partnerID The ID of the partner we are building this socket for
		*/
		public SocketThread(Socket s, int partnerID) throws IOException
		{
			connections[partnerID] = this;
			initialize(s);
		}
		private void initialize(Socket s) throws IOException
		{
			this.s = s;
			// create read-write buffer (ACSII)
			sout = new ObjectOutputStream(s.getOutputStream());
			sin = new ObjectInputStream(s.getInputStream());
			running = true;
			this.start();
		}
		/**
		* Constantly queries the TCP socket for incoming info, if there is nothing there
		* it blocks
		*/
		public void run()
		{
			////Message msg = null;
			while(running)
			{
				Message msg = null;	// renew 'msg' on every loop
				
				try
				{
					msg = ( Message )sin.readObject();
				}
				catch(SocketException e)
				{
					//This is normal behavior when a connection is closed.
					Utils.print_log("Socket Closed");
					//*****We can use this to detect Node death!!!!!*******
					running = false;
				}
				catch(EOFException e){
					System.out.println("Connection lost");
				}
				catch(IOException e)
				{
					System.out.println("Error reading a new message: " + e);
				}
				catch(ClassNotFoundException e)
				{
					System.out.println("Could not load Message Class, make sure Message.class is in this directory");
					System.exit(0);
				}
				if(msg != null)
				{
					// a msg arrives to the port
					receive(msg);
				}else{
					Utils.print_log( "i ain't got nothing" );
				}
				
				// to give some period
				try{
					sleep( 0, 100 );
				}catch( InterruptedException e ){
					e.printStackTrace();
				}
			}
		}
		/**
		* Writes a message to the TCP socket.
		* @param msg The outgoing message to place on the socket
		*/
		public boolean send(Message msg)
		{
			try
			{
				sout.writeObject(msg);
			}
			catch(Exception e)
			{
				return false;
			}
			return true;
		}
		
		/**
		* Close down this socket and the listening Thread
		*/
		public void close() throws IOException
		{
			running = false;
			s.close();
		}
		public String toString()
		{
			return s.toString();
		}
	}
}

/* Todo
 * - handling of disconnected node (addConnection should be attempted for these nodes)
 * - the msg handling routine - Ping / drop / kill stuff
 * - Complete is_this_everyone()
 */